package com.atcyj.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atcyj.gamll.realtime.common.bean.TableProcessDim;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.HBaseUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;

/**
 * @author: cyj
 * @date: 2024/8/15
 */
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        JSONObject jsonObject = value.f0;
        TableProcessDim dim = value.f1;

        String type = jsonObject.getString("type");
        JSONObject data = jsonObject.getJSONObject("data");
        if ("delete".equals(type)) {
            // 删除一整行数据
            delete(data, dim);
        } else {
            // 覆盖写入
            put(data, dim);

        }

    }

    private void delete(JSONObject data, TableProcessDim dim) throws IOException {

        String rowKey = data.getString(dim.getSinkRowKey());
        HBaseUtil.deleteRow(connection, Constant.HBASE_NAMESPACE, dim.getSinkTable(), rowKey);
    }

    private void put(JSONObject data, TableProcessDim dim) throws IOException {
        String rowKey = data.getString(dim.getSinkRowKey());
        HBaseUtil.putRow(connection, Constant.HBASE_NAMESPACE, dim.getSinkTable(), rowKey, dim.getSinkFamily(), data);
    }
}
